package com.requestfactory.wire.spi.http;

import java.io.IOException;
import java.io.OutputStream;
import java.io.Serializable;
import java.lang.reflect.Method;
import java.net.InetSocketAddress;
import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.logging.Level;
import java.util.logging.Logger;

import javax.crypto.CipherInputStream;

import org.apache.commons.httpclient.HttpStatus;
import org.apache.commons.lang.SerializationUtils;

import com.requestfactory.wire.WireError;
import com.requestfactory.wire.WireRuntimeException;
import com.requestfactory.wire.WireSkeletonException;
import com.requestfactory.wire.spi.WireSPIRequest;
import com.requestfactory.wire.spi.WireSPIResponse;
import com.requestfactory.wire.spi.WireSPIServerId;
import com.requestfactory.wire.spi.WireSPISkeleton;
import com.sun.net.httpserver.HttpExchange;
import com.sun.net.httpserver.HttpHandler;
import com.sun.net.httpserver.HttpServer;

public class WireSPISkeletonSunHTTPServerImpl extends WireSPISkeleton {
	
	private static final Logger logger = 
		Logger.getLogger(WireSPISkeletonSunHTTPServerImpl.class.getName());
	
	private final Map<String,ServiceImplWrapper> serviceMap = 
		new HashMap<String,ServiceImplWrapper>();
	
	private final HttpServer server;
	private final String wireSkeletonURL;
	private final AtomicLong serverInvokations; 
	private final AtomicInteger serverConcurrentInvokations; 

	public WireSPISkeletonSunHTTPServerImpl(
			WireSPIServerId wireSPIServerId) 
	throws WireSkeletonException {
		super(wireSPIServerId);
		InetSocketAddress addr = new InetSocketAddress(
				wireSPIServerId.getPort());
		this.wireSkeletonURL = 
			"http://" + addr.getAddress().getHostAddress() + 
			":" + wireSPIServerId.getPort() + wireSPIServerId.getContext(); 
		try {
			//
			// http://www.java2s.com/Code/Java/JDK-6/LightweightHTTPServer.htm
			//
			this.server = HttpServer.create(addr, 0);
			server.createContext(
					wireSPIServerId.getContext(), 
					new InternalHttpHandler());
			server.setExecutor(Executors.newCachedThreadPool());
			server.start();
			if(logger.isLoggable(Level.INFO)) {
				logger.log(Level.INFO, 
						"WIRE-HTTP-SKELETON server is listening on " + 
						this.wireSkeletonURL);
			}
		} catch(IOException ex) {
			throw new WireSkeletonException(
					WireError.WIRE_SKELETON_CANNOT_CREATE_SERVER_INSTANCE, 
					"Cannot create skeleton server instance on '" +
					this.getWireSPIServerId().toString() + 
					"', see nested cause exception",
					ex);
		}
		this.serverInvokations = new AtomicLong();
		this.serverConcurrentInvokations = new AtomicInteger();
	}
	
	@Override
	public String getWireSkeletonURL() {
		return this.wireSkeletonURL;
	}

	@Override
	public synchronized void bindService(
			Class<?> serviceInterfaceClass, 
			Object serviceImplementation) 
	throws WireSkeletonException {
		assert serviceInterfaceClass != null;
		assert serviceImplementation != null;
		if(!serviceInterfaceClass.isAssignableFrom(
				serviceImplementation.getClass())) {
			throw new WireSkeletonException(
					WireError.WIRE_SKELETON_INTERFACE_NOT_INSTANCEOF_IMPL,
					"Service implementation instance '" + 
					serviceImplementation.getClass().getName() + 
					"' in not instace of service interface '" +
					serviceInterfaceClass.getName() + 
					"', check input arguments");
		}
		assert serviceInterfaceClass.isAssignableFrom(
				serviceImplementation.getClass());
		if(serviceMap.containsKey(serviceInterfaceClass.getName())) {
			throw new WireSkeletonException(
					WireError.WIRE_SKELETON_CANNOT_REGISTER_DUPLICATE_INTERFACE,
					"Service interface '" + 
					serviceInterfaceClass.getName() + 
					"' already registered in this skeleton server '" +
					this.wireSkeletonURL + "'");
		}
		serviceMap.put(serviceInterfaceClass.getName(), 
				new ServiceImplWrapper(serviceImplementation));
	}

	private class InternalHttpHandler implements HttpHandler {
		
		private InternalHttpHandler() {
		}
		
		public void handle(HttpExchange exchange) 
		throws IOException {

			ExchangeStatistics stats = new ExchangeStatistics();
			if("POST".equalsIgnoreCase(exchange.getRequestMethod())) {
				stats.beginNanos = System.nanoTime();
				//
				// deserialize request
				//
				WireSPIRequest request = null;
				try {
					long beginDeserializeNanos = System.nanoTime();
					request = (WireSPIRequest)
					SerializationUtils.deserialize(
							new CipherInputStream(exchange.getRequestBody(), 
									wireSPIServerId.getCipher()));
					long endDeserializeNanos = System.nanoTime();
					stats.deserializeRequestElapsedNanos = 
						endDeserializeNanos - beginDeserializeNanos;  
				} catch(Throwable ex) {
					writeResponse(exchange, null, new WireSPIResponse(
							new WireRuntimeException(
									WireError.WIRE_SPI_CANNOT_DESERIALIZE_REQUEST,
									"Cannot deserialize request")), stats);
					return;
				}
				assert request != null;
				//
				// handle request
				//
				handlePost(exchange, request, stats);
				return;
			}
			writeResponse(exchange, null, new WireSPIResponse( 
					new WireRuntimeException(
							WireError.WIRE_SPI_UNSUPPORTED_HTTP_METHOD,
							"Received unsupported request method '" + 
							exchange.getRequestMethod() +	
							"', only POST method supported")), 
							stats);
		}
		
		private void handlePost(
				HttpExchange exchange,
				WireSPIRequest request,
				ExchangeStatistics stats) 
		throws IOException {
			assert exchange != null;
			assert request != null;
			assert stats != null;
			Serializable responseValue = null;
			ServiceImplWrapper serviceImplWrapper = null; 
			try {
				//
				// lookups service by class name
				//
				serviceImplWrapper = serviceMap.get(request.getClassName());
				if(serviceImplWrapper == null) {
					//
					// fatal error: no such service registered in this skeleton
					//
					writeResponse(exchange,
							new WireSPIResponse(
							new WireRuntimeException(
									WireError.WIRE_SPI_CANNOT_LOOKUP_SERVICE,
									"No such service '" +
									request.getClassName() +
									"' registered in skeleton server '" +
									wireSkeletonURL + "'")), stats);
					return;
				}
				assert serviceImplWrapper != null;
				if(logger.isLoggable(Level.INFO)) {
					StringBuilder sb = new StringBuilder();
					sb.append("WIRE-HTTP-SKELETON lookup method '");
					methodSignature(sb, request, serviceImplWrapper.service);
					sb.append("' on '" + wireSkeletonURL + "'");
					logger.log(Level.INFO, sb.toString());
				}
				//
				// lookup service method
				//
				Method method = null; 
				try {
					method = serviceImplWrapper.service.getClass().getMethod(
							request.getMethodName(), 
							request.getParameterTypes());
				} catch(Throwable ex) {
					writeResponse(exchange,	serviceImplWrapper, 
							new WireSPIResponse(
							new WireRuntimeException(
									WireError.WIRE_SPI_CANNOT_LOOKUP_METHOD_SIGNATURE,
									methodSignatureNotFound(new StringBuilder(),
											request, 
											serviceImplWrapper.service).toString())), 
											stats);
				}
				if(method == null) {
					//
					// fatal error - no such method signature in service
					//
					writeResponse(exchange, serviceImplWrapper,
							new WireSPIResponse(
							new WireRuntimeException(
									WireError.WIRE_SPI_CANNOT_LOOKUP_METHOD_SIGNATURE,
									methodSignatureNotFound(new StringBuilder(),
											request, 
											serviceImplWrapper.service).toString())), 
											stats);
					return;
				}
				assert method != null;
				//
				// invoke service method
				//
				long beginMethodInvokeNanos = System.nanoTime();
				serverInvokations.incrementAndGet();
				serverConcurrentInvokations.incrementAndGet();
				serviceImplWrapper.serviceConcurrentInvokations.incrementAndGet();
				serviceImplWrapper.serviceInvokations.incrementAndGet();
				try {
					responseValue = 
						(Serializable)method.invoke(
								serviceImplWrapper.service, 
								(Object[])request.getArguments());
				} finally {
					long endMethodInvokeNanos = System.nanoTime();
					stats.methodInvokeElapsedNanos = 
						endMethodInvokeNanos - beginMethodInvokeNanos;
					serviceImplWrapper.serviceConcurrentInvokations.decrementAndGet();
					serverConcurrentInvokations.decrementAndGet();
				}
			} catch(Throwable ex) {
				//
				// service thru exception
				//
				responseValue = ex;
			}
			writeResponse(exchange, serviceImplWrapper, 
					new WireSPIResponse(responseValue), stats);
		}

		private void writeResponse(
				HttpExchange exchange,
				WireSPIResponse response,
				ExchangeStatistics stats) 
		throws IOException {
			this.writeResponse(exchange, null, response, stats); 
		}
		
		private void writeResponse(
				HttpExchange exchange,
				ServiceImplWrapper serviceImplWrapper,
				WireSPIResponse response,
				ExchangeStatistics stats) 
		throws IOException {
			assert exchange != null;
			assert response != null;
			assert stats != null;
			exchange.sendResponseHeaders(HttpStatus.SC_OK, 0);
			OutputStream responseBody = exchange.getResponseBody();
			exchange.getResponseHeaders().set(
					"Content-Type", "application/binary");
			exchange.getResponseHeaders().set(
					"Content-Length", "-1");
			long beginDeserializeNanos = System.nanoTime();
			SerializationUtils.serialize(
					response, responseBody);
			long endDeserializeNanos = System.nanoTime();
			stats.serializeResponseElapsedNanos = 
				endDeserializeNanos - beginDeserializeNanos;  
			responseBody.close();
			long endNanos = System.nanoTime();
			stats.elapsedNanos = endNanos - stats.beginNanos;
			if(logger.isLoggable(Level.INFO)) {
				logger.log(Level.INFO, 
						"WIRE-HTTP-SKELETON invoke  statistics " + stats);
			}
			//
			// record statistics only for normal responses
			// skip exception responses
			//
			if(serviceImplWrapper != null &&
					!(response.getValue() instanceof Throwable)) {
				serviceImplWrapper.put(stats);
				if(logger.isLoggable(Level.INFO)) {
					logger.log(Level.INFO, 
							"WIRE-HTTP-SKELETON average statistics " + 
							serviceImplWrapper.average());
					logger.log(Level.INFO, 
							"WIRE-HTTP-SKELETON serviceConcurrentInvokations=" + 
							serviceImplWrapper.serviceConcurrentInvokations);
					logger.log(Level.INFO, 
							"WIRE-HTTP-SKELETON serverConcurrentInvokations=" + 
							serverConcurrentInvokations);
				}
			}
		}
		
		private StringBuilder methodSignature(StringBuilder sb,
				WireSPIRequest request, Object service) {
			assert sb != null;
			assert request != null;
			assert service != null;
			sb.append(service.getClass().getSimpleName());
			sb.append(".").append(request.getMethodName());
			sb.append("(");
			int i = 0;
			for(Class<?> parameterType : request.getParameterTypes()) {
				if(i++ > 0) sb.append(",");
				sb.append(parameterType.getSimpleName());
			}
			sb.append(")");
			return sb;
		}
		
		private StringBuilder methodSignatureNotFound(StringBuilder sb,
				WireSPIRequest request, Object service) {
			assert sb != null;
			assert request != null;
			assert service != null;
			sb.append("No such method signature '");
			methodSignature(sb, request, service);
			sb.append("' found in service '");
			sb.append(request.getClassName());
			sb.append("' in skeleton server '");
			sb.append(wireSkeletonURL).append("'");
			return sb;
		}
	}
	
	private static class ExchangeStatistics implements Serializable {
		private static final long serialVersionUID = 20080926000905L;
		long beginNanos = 0L;
		//
		// MulticastSocket, DatagramPacket
		//
		long elapsedNanos = 0L;		
		long deserializeRequestElapsedNanos = 0L;
		long methodInvokeElapsedNanos = 0L;
		long serializeResponseElapsedNanos = 0L;
		@Override
		public String toString() {
			return "\n\telapsedNanos=" + elapsedNanos +
			"\n\t\tdeserializeRequestElapsedNanos=" + deserializeRequestElapsedNanos +
			"\n\t\tmethodInvokeElapsedNanos=" + methodInvokeElapsedNanos +
			"\n\t\tserializeResponseElapsedNanos=" + serializeResponseElapsedNanos;
		}
		
	}
	
	private static class ServiceImplWrapper {
		
		private static final int MAX_CAPACITY = 10;
		
		private final Object service;
		private final Map<Long,ExchangeStatistics> statsMap;
		private final AtomicLong serviceInvokations; 
		private final AtomicInteger serviceConcurrentInvokations; 
		
		ServiceImplWrapper(Object service) {
			assert service != null;
			this.service = service;
			this.statsMap = Collections.synchronizedMap( 
				new LRUCache<Long,ExchangeStatistics>(MAX_CAPACITY));
			this.serviceInvokations = new AtomicLong();
			this.serviceConcurrentInvokations = new AtomicInteger();
		}
		
		synchronized void put(ExchangeStatistics stats) {
			assert stats != null;
			statsMap.put(System.currentTimeMillis(), stats);
		}
		
		synchronized ExchangeStatistics average() {
			ExchangeStatistics average = new ExchangeStatistics();
			int size = statsMap.size();			
			for(Map.Entry<Long,ExchangeStatistics> entry : statsMap.entrySet()) {
				average.elapsedNanos += 
					entry.getValue().elapsedNanos;
				average.deserializeRequestElapsedNanos += 
					entry.getValue().deserializeRequestElapsedNanos;
				average.methodInvokeElapsedNanos +=
					entry.getValue().methodInvokeElapsedNanos;
				average.serializeResponseElapsedNanos +=
					entry.getValue().serializeResponseElapsedNanos;
			}
			average.elapsedNanos /= size;
			average.deserializeRequestElapsedNanos /= size; 
			average.methodInvokeElapsedNanos /= size;
			average.serializeResponseElapsedNanos /= size;
			return average;
		}
	}
	
	private static class LRUCache<K,V> extends LinkedHashMap<K,V> {
		
		private static final long serialVersionUID = 1L;
		
		private final int maxCapacity;
		
		LRUCache(int maxCapacity) {
			super(maxCapacity);
			this.maxCapacity = maxCapacity;
		}

		@Override
		protected boolean removeEldestEntry(Entry<K,V> eldest) {
			return size() > maxCapacity;
		}
		
	}
}
